import org.apache.commons.net.ntp.TimeStamp;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.concurrent.atomic.AtomicReference;

public class HbaseOP {
    private static final int DEFAULT_VERSION = 3;
    public Configuration configuration; // 管理Hbase的配置信息
    public Connection connection; // 管理Hbase连接
    public Admin admin; // 管理Hbase数据库的信息
    public Scan scan;
    Table table;

    public static int getDefaultVersion() {
        return DEFAULT_VERSION;
    }

    public HbaseOP() {
        init();
    }

    public HbaseOP(String tableName) {
        init();
        try {
            this.table = connection.getTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void init(){
        scan=new Scan();
        configuration = HBaseConfiguration.create();
//        configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");//主节点
        configuration.set("hbase.zookeeper.quorum", "bink.website");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            System.out.println("connected...");
            admin = connection.getAdmin();
        }catch(IOException e){
            e.printStackTrace();
        }

    }

    public void createTable(String myTableName,String[] colFamily) throws IOException{
        TableName tableName = TableName.valueOf(myTableName);
        if(admin.tableExists(tableName)){
            System.out.println("Table "+myTableName+" exists");
        }else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
//            String version = hTableDescriptor.getConfiguration().put("VERSION", DEFAULT_VERSION);
//            hTableDescriptor.setConfiguration("VERSION",DEFAULT_VERSION);
//            hTableDescriptor.getConfiguration().forEach((k,v)-> System.out.println(k+":"+v));
            for(String str:colFamily){
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(str);
                hColumnDescriptor.setMaxVersions(DEFAULT_VERSION);
                hTableDescriptor.addFamily(hColumnDescriptor);
//                System.out.println(str);
            }
            admin.createTable(hTableDescriptor);
            System.out.println("Table "+myTableName+" created...");
        }
    }

    // 添加单元格数据
    public void insertData(String tableName,String rowKey,String colFamily,String col,String val,long timestamp) throws IOException{
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(rowKey.getBytes());
        put.addColumn(colFamily.getBytes(),col.getBytes(),val.getBytes());
        put.setTimestamp(timestamp);
        table.put(put);
        table.close();
    }

    public void insertData(String tableName,String rowKey,String colFamily,String col,String val) throws IOException{
        insertData(tableName, rowKey, colFamily, col, val);
    }

    //浏览数据
    public void getData(String tableName,String rowKey,String colFamily,String col) throws IOException{
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(rowKey.getBytes());
        get.addColumn(colFamily.getBytes(),col.getBytes());
        Result result =table.get(get);
        System.out.println(new String(result.getValue(colFamily.getBytes(),col==null?null:col.getBytes())));
        table.close();
    }


    // 操作数据库之后，关闭连接
    public void close(){
        try{
            if(admin!=null){
                admin.close(); // 退出用户
            }
            if(null != connection){
                connection.close(); // 关闭连接
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }


    //删除表
    public void deleteTable(String tableName){

        try {
            TableName tablename = TableName.valueOf(tableName);
            admin = connection.getAdmin();
            if(admin.tableExists(tablename)) {
                if(!admin.isTableDisabled(tablename)) {
                    admin.disableTable(tablename);
                }
                admin.deleteTable(tablename);
            }
            System.out.println("Table "+tableName+" deleted...");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public void scanTable(String tableName) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        ResultScanner scanner = table.getScanner(scan);
//        for (Result res : scanner) {
//            System.out.println(res);
//        }
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(
                        "行键:" + new String(CellUtil.cloneRow(cell)) + "\t" +
                                "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +
                                "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +
                                "值:" + new String(CellUtil.cloneValue(cell)) + "\t" +
                                "时间戳:" + cell.getTimestamp());
            }
        }
        scanner.close();
    }

    public void truncateTable(String tableName) throws IOException {
        TableName tablename = TableName.valueOf(tableName);
        if(admin.isTableEnabled(tablename)) {
            admin.disableTable(tablename);
        }
        admin.truncateTable(tablename,true);
    }

    public int countTableRows(String tableName) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        ResultScanner scanner = table.getScanner(new Scan());
//        for (Result res : scanner) {
//            System.out.println(res);
//        }
//        for (Result result : scanner) {
//            for (Cell cell : result.rawCells()) {
//                System.out.println(
//                        "行键:" + new String(CellUtil.cloneRow(cell)) + "\t" +
//                                "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +
//                                "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +
//                                "值:" + new String(CellUtil.cloneValue(cell)) + "\t" +
//                                "时间戳:" + cell.getTimestamp());
//            }
//        }
        AtomicReference<Integer> count = new AtomicReference<>(0);
        for (Result res : scanner) {
            count.set(count.get()+1);
        }
        scanner.close();
        return count.get();
    }
    //End

}